package com.pt25.base.dep.es;

import com.pt25.base.constants.BaseConstants;
import com.pt25.base.dep.es.ddo.Mapping;
import com.pt25.base.util.CollectionUtil;
import jakarta.annotation.Resource;
import org.apache.http.HttpHost;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.settings.put.UpdateSettingsRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.PutMappingRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Component
public class Es7Service {
    @Resource(name = "restHighLevelClientMap")
    private Map<String, RestHighLevelClient> restHighLevelClientMap;

    public RestHighLevelClient getRestHighLevelClient() {
        return restHighLevelClientMap.get(BaseConstants.DEFAULT);
    }

    public RestHighLevelClient getRestHighLevelClient(String name) {
        return restHighLevelClientMap.get(name);
    }


    // 创建索引
    public void createIndex(String clusterName, String indexName) throws IOException {
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        getRestHighLevelClient(clusterName).indices().create(request, RequestOptions.DEFAULT);
    }

    //创建索引
    public void createIndex(String cluster, Mapping mapping) throws IOException {
        if (!existIndex(cluster, mapping.getIndex())) {
            // 创建索引请求
            CreateIndexRequest request = new CreateIndexRequest(mapping.getIndex());
            // 设置分片数和副本数
            request.settings(Settings.builder()
                    .put("index.number_of_shards", mapping.getShards())
                    .put("index.number_of_replicas", mapping.getReplicas())
            );
            // 执行创建索引请求
            getRestHighLevelClient(cluster).indices().create(request, RequestOptions.DEFAULT);
            // 更新索引结构
            updateIndex(cluster, mapping);
        }
    }

    //更新索引
    public void updateIndex(String cluster, Mapping mapping) throws IOException {
        if (existIndex(cluster, mapping.getIndex())) {
            // 构建索引结构
            if (CollectionUtil.isNotEmpty(mapping.getProperties())) {
                PutMappingRequest putMappingRequest = new PutMappingRequest(mapping.getIndex());
                putMappingRequest.source(mapping.createMapping(), XContentType.JSON);
                getRestHighLevelClient(cluster).indices().putMapping(putMappingRequest, RequestOptions.DEFAULT);
            }
            // 设置新的副本数
            UpdateSettingsRequest request = new UpdateSettingsRequest(mapping.getIndex());
            Settings settings = Settings.builder()
                    .put("index.number_of_replicas", mapping.getReplicas())
                    .build();
            request.settings(settings);
            getRestHighLevelClient(cluster).indices().putSettings(request, RequestOptions.DEFAULT);
        }
    }


    //删除索引
    public void deleteIndex(String cluster, String index) throws IOException {
        if (existIndex(cluster, index)) {
            // 定义删除索引请求
            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(index);
            getRestHighLevelClient(cluster).indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
        }
    }


    //判断索引是否存在
    public Boolean existIndex(String cluster, String index) throws IOException {
        // 定义 GetIndexRequest
        GetIndexRequest request = new GetIndexRequest(index);
        return getRestHighLevelClient(cluster).indices().exists(request, RequestOptions.DEFAULT);
    }


    // 索引文档
    public IndexResponse indexDocument(String cluster, String index, String id, Map<String, Object> document) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        IndexResponse response = client.index(new IndexRequest(index)
                .id(id)
                .source(document, XContentType.JSON), RequestOptions.DEFAULT);
        return response;
    }

    // 获取文档
    public GetResponse getDocument(String cluster, String index, String id) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        GetResponse response = client.get(new GetRequest(index, id), RequestOptions.DEFAULT);
        return response;
    }

    // 更新文档
    public UpdateResponse updateDocument(String cluster, String index, String id, Map<String, Object> update) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        UpdateRequest request = new UpdateRequest(index, id).doc(update);
        UpdateResponse response = client.update(request, RequestOptions.DEFAULT);
        return response;
    }

    // 删除文档
    public DeleteResponse deleteDocument(String cluster, String index, String id) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        DeleteResponse response = client.delete(new DeleteRequest(index, id), RequestOptions.DEFAULT);
        return response;
    }

    // 搜索文档
    public SearchResponse searchDocuments(String cluster, String index, String field, String value) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchQuery(field, value));
        SearchRequest request = new SearchRequest(index);
        request.source(sourceBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        return response;
    }

    // 批量操作
    public BulkResponse bulkOperations(String cluster, BulkRequest bulkRequest) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        return client.bulk(bulkRequest, RequestOptions.DEFAULT);
    }

    // 分页查询
    public SearchResponse searchDocumentsWithPagination(String cluster, String index, QueryBuilder query, int page, int size) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(query);
        sourceBuilder.from((page - 1) * size);
        sourceBuilder.size(size);
        SearchRequest request = new SearchRequest(index);
        request.source(sourceBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        return response;
    }

    // 批量更新
    public BulkByScrollResponse updateByQuery(String cluster, UpdateByQueryRequest request) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        return client.updateByQuery(request, RequestOptions.DEFAULT);
    }

    // 批量删除
    public BulkByScrollResponse deleteByQuery(String cluster, DeleteByQueryRequest request) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        return client.deleteByQuery(request, RequestOptions.DEFAULT);
    }

    // 使用通用查询
    public SearchResponse searchByQuery(String cluster, String index, QueryBuilder query) throws IOException {
        RestHighLevelClient client = getRestHighLevelClient(cluster);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(query);
        SearchRequest request = new SearchRequest(index);
        request.source(sourceBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        return response;
    }
    /**
     * 基于id批量查询数据
     * @param cluster es集群名称
     * @param index   索引名称
     * @param ids     文档id
     * @return
     * @throws IOException
     */
    public List<GetResponse> searchDocumentsByIds(String cluster, String index, List<String> ids) throws IOException {
        List<GetResponse> getResponses = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(ids)) {
            RestHighLevelClient client = getRestHighLevelClient(cluster);
            // 创建MultiGetRequest
            MultiGetRequest request = new MultiGetRequest();
            // 添加你要查询的index和id
            for (String id : ids) {
                request.add(new MultiGetRequest.Item(index, id));
            }
            // 执行批量查询
            MultiGetResponse multiGetResponse = client.mget(request, RequestOptions.DEFAULT);
            MultiGetItemResponse[] multiGetItemResponses = multiGetResponse.getResponses();
            for (MultiGetItemResponse multiGetItemResponse : multiGetItemResponses) {
                if (!multiGetItemResponse.isFailed()) {
                    GetResponse response = multiGetItemResponse.getResponse();
                    getResponses.add(response);
                }
            }
        }

        return getResponses;
    }

    /**
     * 复制索引
     *
     * @param sourceUrl        源地址，有可能是集群，要把每个地址都加入 每个url应该是：地址 + 端口号，例如：127.0.0.1:9200
     * @param destinationUrl   目标地址
     * @param sourceIndex      源索引
     * @param destinationIndex 目标索引
     */
    public void copySourceIndexToDestinationIndex(List<String> sourceUrl, List<String> destinationUrl, String sourceIndex, String destinationIndex) {
        if (null == sourceUrl || null == destinationUrl || CollectionUtil.isEmpty(sourceUrl) || CollectionUtil.isEmpty(destinationUrl)) {
            return;
        }
        RestHighLevelClient sourceClient = createRestHighLevelClient(sourceUrl);
        RestHighLevelClient destinationClient = createRestHighLevelClient(destinationUrl);
        try {
            copy(sourceClient, destinationClient, sourceIndex, destinationIndex);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                sourceClient.close();
                destinationClient.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private RestHighLevelClient createRestHighLevelClient(List<String> urls) {
        List<HttpHost> httpHosts = new ArrayList<>();
        for (String url : urls) {
            String[] parts = url.split(":");
            if (parts.length == 2) {
                httpHosts.add(new HttpHost(parts[0], Integer.parseInt(parts[1]), "http"));
            }
        }
        return new RestHighLevelClient(RestClient.builder(httpHosts.toArray(new HttpHost[0])));
    }

    private void copy(RestHighLevelClient sourceClient, RestHighLevelClient destinationClient, String sourceIndex, String destinationIndex) {
        try {

            final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
            SearchRequest searchRequest = new SearchRequest(sourceIndex);
            searchRequest.scroll(scroll);
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.query(QueryBuilders.matchAllQuery());
            searchRequest.source(searchSourceBuilder);

            SearchResponse searchResponse = sourceClient.search(searchRequest, RequestOptions.DEFAULT);
            String scrollId = searchResponse.getScrollId();
            boolean hasMoreHits = searchResponse.getHits().getHits().length > 0;

            while (hasMoreHits) {
                BulkRequest bulkRequest = new BulkRequest();
                for (SearchHit hit : searchResponse.getHits().getHits()) {
                    String docId = hit.getId();
                    IndexRequest indexRequest = new IndexRequest(destinationIndex)
                            .id(docId)
                            .source(hit.getSourceAsString(), XContentType.JSON);
                    bulkRequest.add(indexRequest);
                }

                if (bulkRequest.numberOfActions() > 0) {
                    BulkResponse bulkResponse = destinationClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                    if (bulkResponse.hasFailures()) {
                        // Handle failures
                        System.out.println("Bulk indexing had failures: " + bulkResponse.buildFailureMessage());
                    }
                }

                SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                scrollRequest.scroll(scroll);
                searchResponse = sourceClient.scroll(scrollRequest, RequestOptions.DEFAULT);
                scrollId = searchResponse.getScrollId();
                hasMoreHits = searchResponse.getHits().getHits().length > 0;
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
